package com.navinfo.tripanalysis.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.VehicleBusinessInfo;
import com.navinfo.tripanalysis.common.arithmetic.common.VehicleInfo;
import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.service.AbstractLoadVehicleInfoService;
import org.apache.commons.io.FileUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.File;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;
import java.util.stream.StreamSupport;

/**
 * 加载东风车辆的配置信息
 * @author web
 */
public class LoadVehicleInfoServiceImpl_DF extends AbstractLoadVehicleInfoService {

    @Override
    protected String getTable() {
        return  "(SELECT terminal.t_sim,car.car_model_code " +
                    "FROM hy_car car,hy_terminal terminal " +
                    "WHERE car.CAR_TERMINAL=terminal.t_id) t";
    }


    private String vehicle_hql;

    @Override
    public String getVehicle_hql() {
        return vehicle_hql;
    }

    @Override
    public void setVehicle_hql(String vehicle_hql) {
        this.vehicle_hql = vehicle_hql;
    }

    @Override
    protected JavaPairRDD<Long, VehicleInfo> getVehicleInfoPair(SparkSession session, JavaRDD<Row> rowJavaRDD) {
        Broadcast<Map<String, String>> broadcast = new JavaSparkContext(session.sparkContext()).broadcast(getWeightLoad());
        Map<Long,VehicleBusinessInfo> vehicleBusinessInfoMap = new HashMap<>();
        vehicleBusinessInfoMap.putAll(getBusinessInfo(session).collectAsMap());
//        vehicleBusinessInfoMap = getBusinessInfo(session).collectAsMap();
        Broadcast<Map<Long,VehicleBusinessInfo>> vehicleBusinessInfo =  new JavaSparkContext(session.sparkContext()).broadcast(vehicleBusinessInfoMap);
        JavaPairRDD<Long, VehicleInfo> longVehicleInfoJavaPairRDD = rowJavaRDD.mapPartitionsToPair(rowIterator -> {
            List<Tuple2<Long, VehicleInfo>> vehicleList = new ArrayList<>();
            while (rowIterator.hasNext()) {
                Row row = rowIterator.next();
                String tid = row.getString(0);
                String carModel = row.getString(1);

                VehicleInfo vehicleInfo = new VehicleInfo();

                vehicleInfo.setTerminalId(Long.valueOf(tid));

                if (vehicleBusinessInfo.getValue().get(Long.valueOf(tid)) != null){
                    vehicleInfo.setCarType(Optional.ofNullable(vehicleBusinessInfo.getValue().get(Long.valueOf(tid)).getCarType()).orElse(""));
                    vehicleInfo.setMaxLoad(Optional.ofNullable(vehicleBusinessInfo.getValue().get(Long.valueOf(tid)).getMaxLoad()).orElse(""));

                }

                Map<String, String> weightLoad = broadcast.getValue();
                Optional.ofNullable(weightLoad.get(carModel)).ifPresent(e -> {
                    String[] split = e.split(",");
                    if (null != split && split.length == 3) {
                        vehicleInfo.setWloadNo(carModel);
                        vehicleInfo.setWloadEmpty(Integer.valueOf(split[0]));
                        vehicleInfo.setWloadHalf(Integer.valueOf(split[1]));
                        vehicleInfo.setWloadFull(Integer.valueOf(split[2]));

                    }

                });

                vehicleList.add(new Tuple2(Long.valueOf(tid), vehicleInfo));
            }
            return vehicleList.iterator();
        });
        return longVehicleInfoJavaPairRDD;
    }

    private Map<String, String> getWeightLoad(){
        Map<String, String> result = new HashMap<>();

        List<String> weightLoads = null;
        try {
            File file = FileUtils.getFile( new File(this.getClass().getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile(), "df_weight_load") ;
            weightLoads = FileUtils.readLines(file);
        } catch (URISyntaxException |IOException e) {
            logger.error("加载东风车型载重信息失败...", e);
        }

        if(null != weightLoads){
            weightLoads.forEach(e ->{
                int firstIndex = e.indexOf(",");
                result.put(e.substring(0,firstIndex),  e.substring(firstIndex+1));
            });
        }

        logger.info("加载东风车型载重信息完成，大小{}...",result.size());
        return result;
    }

    private JavaPairRDD<Long, VehicleBusinessInfo> getBusinessInfo(SparkSession session){


        String hql = vehicle_hql;
        JavaPairRDD<Long, VehicleBusinessInfo> vehicleBusinessInfoJavaPairRDD = toJavaRdd(session.sql(hql));


        logger.info("加载东风车型载重信息完成，大小{}...",vehicleBusinessInfoJavaPairRDD.count());
        return vehicleBusinessInfoJavaPairRDD;
    }
    private JavaPairRDD<Long, VehicleBusinessInfo> toJavaRdd(Dataset<Row> dataset) {
       JavaPairRDD<Long, VehicleBusinessInfo> vehicleBusinessInfoJavaPairRDD = dataset.select("tid", "car_type","max_tonnage")
                .toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, VehicleBusinessInfo>) iterator -> {
                    Iterable<Row> iterable = () -> iterator;
                    return StreamSupport.stream(iterable.spliterator(), false)
                            .map(s -> {
                                VehicleBusinessInfo vehicleBusinessInfo = new VehicleBusinessInfo();
                                Long tid = Long.valueOf((String)s.getAs("tid"));
                                String car_type = String.valueOf((Integer)s.getAs("car_type"));
                                String max_tonnage = s.getAs("max_tonnage");
                                vehicleBusinessInfo.setTerminalId(tid);
                                vehicleBusinessInfo.setCarType(car_type);
                                vehicleBusinessInfo.setMaxLoad(max_tonnage);
                                return new Tuple2<>(tid, vehicleBusinessInfo);
                            }).iterator();
                });
      return vehicleBusinessInfoJavaPairRDD;
    }


}
